package com.o2o.cleaning.month.platform.ebusiness_plat.kuaishou.shuang11

import org.apache.spark.sql.SparkSession

object _01_PG_TO_OBS {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .config("spark.debug.maxToStringFields", "500")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.caseSensitive", "true")
      .config("spark.executor.heartbeatInterval", "120s")
      .config("spark.network.timeout", "12000s")
      .config("spark.default.parallelism", "2")
      .config("cluster.name", "O2OElastic")
      .config("spark.sql.crossJoin.enabled", "true")
      .master("local[*]")
      //      .enableHiveSupport()
      .getOrCreate()

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("WARN")

    //    val year = 2021
    val year = 2022
    val month = 1111
    //    val timestamps = Array("20211020", "20211021", "20211022", "20211023", "20211024", "20211025", "20211026", "20211027", "20211028", "20211029", "20211030", "20211031", "20211101", "20211102", "20211103", "20211104", "20211105", "20211106", "20211107", "20211108", "20211109", "20211110", "20211111")
    val timestamps = Array("20221019", "20221020", "20221021", "20221022", "20221023", "20221024", "20221025", "20221026", "20221027", "20221028", "20221029", "20221030", "20221031", "20221101", "20221102", "20221103", "20221104", "20221105", "20221106", "20221107", "20221108", "20221109")
    for (timestamp <- timestamps) {
      println(s"${year}|${month}|${timestamp}")

      val tab = spark.read.format("jdbc")
        .option("driver", "org.postgresql.Driver")
        .option("url", "jdbc:postgresql://114.115.219.204:8000/postgres")
        .option("user", "sysadmin")
        .option("password", "Bigdata@123")
        .option("dbtable",
          s"""
             |(SELECT *,"good_id"*1%20 mo FROM detail."o2o_detail_kuaishou_2022" WHERE timestamp = '${timestamp}'
             |AND price_abnormal_degree < 9 )
             |""".stripMargin)
        .option("partitionColumn", "mo")
        .option("numPartitions", 20)
        .option("lowerBound", 1)
        .option("upperBound", 20)
        .load()

      println(tab.rdd.getNumPartitions)

      tab.repartition(10).write.mode("overwrite")
        //      .orc(s"s3a://dws-data/g_data/kuaishou/oss/${year}/${month}")
        .orc(s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/dws/${year}/${month}/${timestamp}")
    }
    sc.stop()

  }

}
